package com.yubest.mqtt.config;

import com.yubest.mqtt.consts.Consts;
import com.yubest.mqtt.handler.MqttConsumer;
import lombok.Data;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import java.util.UUID;

/**
 * 消费者配置
 *
 * @Author: hweiyu
 * @Date: 2022/11/11 11:50
 */
@Data
@Configuration
@ConfigurationProperties(prefix = "mqtt.consumer")
public class MqttConsumerConfig {

    private String clientId;

    private String[] serverURIs;

    private String username;

    private String password;

    private String defaultTopic;

    @Bean
    public MqttPahoClientFactory consumerFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setServerURIs(serverURIs);
        factory.setConnectionOptions(options);
        return factory;
    }

    @Bean
    public MessageProducer consumerProducer() {
        String id = clientId + "_sub_" + UUID.randomUUID();
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(id, consumerFactory(), defaultTopic);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setOutputChannel(consumerChannel());
        return adapter;
    }

    @Bean(name = Consts.INPUT_CHANNEL)
    public MessageChannel consumerChannel() {
        return new DirectChannel();
    }

    @Bean
    @ServiceActivator(inputChannel = Consts.INPUT_CHANNEL)
    public MessageHandler consumerHandler() {
        return new MqttConsumer();
    }
}
